package cn.doitedu.sql;

import beans.UserAction;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

public class _16_CustomFunction_Scalar {


    public static void main(String[] args) throws Exception {


        // 2024-01-08 21:17:59
        // 有如下输入数据：     1,page_load,1704719879001
        // 要得到如下结果：     1,page_load,1704719879001, 2024-01-08 21:15:00, 2024-01-08 21:10:00,2024-01-08 21:00:00

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        DataStreamSource<String> stream = env.socketTextStream("localhost", 8899);

        SingleOutputStreamOperator<Event> mappedStream = stream.map(new MapFunction<String, Event>() {
            @Override
            public Event map(String line) throws Exception {
                String[] split = line.split(",");
                return new Event(Integer.parseInt(split[0]), split[1], Long.parseLong(split[2]));
            }
        });

        // 流转表
        tenv.createTemporaryView("action_log", mappedStream);  // 把一个javaBean的流，转成表，让它自动推断表结构，需要javaBean是public的
        tenv.executeSql("desc action_log").print();

        //
        // 注册自定义函数
        tenv.createTemporaryFunction("time_trunc", TimeTruncFunction.class);

        tenv.executeSql(
                "select " +
                        "user_id," +
                        "event_id," +
                        "action_time, " +
                        "time_trunc(action_time,5) as  m5_trunc, " +
                        "time_trunc(action_time,10) as m10_trunc," +
                        "time_trunc(action_time,60) as m60_trunc " +
                        "from action_log").print();


        env.execute();
    }

    public static class TimeTruncFunction extends ScalarFunction {

        // 1704719879001
        public String eval(Long action_time, Long intervalMinute) {

            long tmp = intervalMinute * 60 * 1000;
            long truncated = (action_time / tmp) * tmp;

            return DateFormatUtils.format(truncated, "yyyy-MM-dd HH:mm:ss");
        }

    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Event {
        private int user_id;
        private String event_id;
        private long action_time;
    }


}





